package indi.zhifa.recipe.bailan.framework.queue.handler;

import cn.hutool.core.convert.Convert;
import com.alibaba.fastjson2.JSON;
import indi.zhifa.recipe.bailan.framework.common.entity.exception.ServiceException;
import indi.zhifa.recipe.bailan.framework.queue.bean.MsgHandlerMap;
import indi.zhifa.recipe.bailan.framework.queue.entity.EQueueValueType;
import indi.zhifa.recipe.bailan.framework.queue.entity.MsgHandlerInfo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.stream.StreamListener;

import java.util.Map;

@Slf4j
public abstract class BaseRedisMsgConsumerListener implements StreamListener<String, MapRecord<String,Object,byte[]>> {

    public static final String TYPE_KEY = "type";
    public static final String MSG_ID_KEY = "id";
    public static final String DATA_KEY = "data";

    protected final RedisTemplate<String,Object> mRedisTemplate;
    protected final MsgHandlerMap mMsgHandlerMap;
    protected final String mGroup;

    protected BaseRedisMsgConsumerListener(
            String pGroup,
            RedisTemplate<String,Object> pRedisTemplate){
        mMsgHandlerMap = initMsgHandlerMap();
        mRedisTemplate = pRedisTemplate;
        mGroup = pGroup;
    }

    protected abstract MsgHandlerMap initMsgHandlerMap();


    @Override
    public void onMessage(MapRecord<String,Object,byte[]> pMessage){
        RecordId recordId = pMessage.getId();
        String stream = pMessage.getStream();
        Map<Object,byte[]> valueMap = pMessage.getValue();
        Integer typeCode = getInteger(valueMap.get(TYPE_KEY));
        Integer msgId = getInteger(valueMap.get(MSG_ID_KEY));
        byte[] valueByte = valueMap.get(DATA_KEY);
        if(typeCode > EQueueValueType.values().length || typeCode<0){
            throw new ServiceException(TYPE_KEY+"字段缺失或不合法");
        }
        // 查找MsgId的配置
        MsgHandlerInfo msgHandlerInfo = mMsgHandlerMap.getHandler(msgId);
        if(null == msgHandlerInfo){
            return;
        }
        IMessageHandler messageHandler = msgHandlerInfo.getMessageHandler();

        EQueueValueType valueType = EQueueValueType.values()[typeCode];
        try{
            switch (valueType) {
                case DEFAULT:
                case STRING:
                    String strVal = getStr(valueByte);
                    messageHandler.handler(strVal);
                    break;
                case JSON:
                    Class cls = msgHandlerInfo.getCls();
                    Object obj = null;
                    if(null != cls){
                        obj = getObject(valueByte,cls);
                    }else{
                        obj = getObject(valueByte,Object.class);
                    }
                    messageHandler.handler(obj);
                    break;
                case INT:
                    Integer intVal = getInteger(valueByte);
                    messageHandler.handler(intVal);
                    break;
                case LONG:
                    Long longVal = getLong(valueByte);
                    messageHandler.handler(longVal);
                    break;
            }
            mRedisTemplate.opsForStream().acknowledge(mGroup,pMessage);
        }catch (Exception ex){
            onError(recordId.getValue(),stream,msgId,valueByte,ex);
        }
    }

    protected  void onError(String pId, String pStream, Integer pMsgId, byte[] pValueByte, Exception ex){
        log.error("stream:{}-id:{}-msgId:{} 解析发生错误，错误信息是{}",pId,pStream,pMsgId,ex.getMessage());
    }

    protected String getStr(byte[] pByte){
        return new String(pByte);
    }

    protected Integer getInteger(byte[] pByte){
        return Convert.bytesToInt(pByte);
    }

    protected Long getLong(byte[] pByte){
        return Convert.bytesToLong(pByte);
    }

    protected <T> T getObject(byte[] pByte, Class<T> pCls){
        T tObj = JSON.parseObject(pByte,pCls);
        return tObj;
    }


}
